package com.tencent.cloud.iov.util.rx;

import com.tencent.mars.xlog.Log;

import io.reactivex.Notification;
import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.ObservableTransformer;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.subjects.BehaviorSubject;

/**
 * 过滤订阅前的错误信息
 *
 * @author showxiao
 * @date 2018/9/7
 */
public class OnErrorCacheOnSubscribe<T> implements ObservableTransformer<Notification<T>, T> {

    private static final String TAG = "OnErrorCacheOnSubscribe";

    private final BehaviorSubject<T> mCacheSubject;
    private boolean mIsErrorOnSubscribe = false;

    public OnErrorCacheOnSubscribe(BehaviorSubject<T> cacheSubject) {
        mCacheSubject = cacheSubject;
    }

    @Override
    public ObservableSource<T> apply(final Observable<Notification<T>> upstream) {
        return upstream
                .doOnSubscribe(new Consumer<Disposable>() {
                    @Override
                    public void accept(Disposable disposable) throws Exception {

                        // 判断订阅前是否有错误信息
                        if (upstream instanceof BehaviorSubject) {
                            BehaviorSubject<Notification<T>> subject =
                                    (BehaviorSubject<Notification<T>>) upstream;
                            mIsErrorOnSubscribe =
                                    subject.hasValue() && subject.getValue().isOnError();
                        }

                        Log.d(TAG, "OnErrorCacheOnSubscribe doOnSubscribe: mIsErrorOnSubscribe = "
                                        + mIsErrorOnSubscribe);
                    }
                })
                .map(new Function<Notification<T>, T>() {
                    @Override
                    public T apply(Notification<T> notification) throws Exception {
                        if (notification.isOnNext()) {
                            mIsErrorOnSubscribe = false;
                            return notification.getValue();
                        }

                        // 忽略订阅前的请求错误
                        if (mIsErrorOnSubscribe) {
                            mIsErrorOnSubscribe = false;

                            // 发射最新的缓存有效数据
                            if (mCacheSubject != null && mCacheSubject.hasValue()) {
                                return mCacheSubject.getValue();
                            }

                            throw new WrappedException();
                        }

                        throw new WrappedException(notification.getError());

                    }
                })
                .onErrorResumeNext(new Function<Throwable, ObservableSource<? extends T>>() {
                    @Override
                    public ObservableSource<? extends T> apply(Throwable throwable) throws Exception {
                        if (throwable instanceof WrappedException) {
                            Throwable actual = ((WrappedException) throwable).getActual();
                            if (actual == null) {
                                return Observable.empty();
                            }
                            return Observable.error(actual);
                        }

                        return Observable.error(throwable);
                    }
                });
    }

    private static class WrappedException extends RuntimeException {

        private Throwable mActual;

        public WrappedException() {
            this(null);
        }

        public WrappedException(Throwable actual) {
            mActual = actual;
        }

        public Throwable getActual() {
            return mActual;
        }
    }
}
